package com.group1.realtime.app.dwd.db;

import com.group1.realtime.app.BaseSQLApp;
import com.group1.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import static com.group1.realtime.common.Constant.*;

public class DwdExamAttend extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdExamAttend().init(
                4083,
                2,
                "DwdExamAttend"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"DwdExamAttend");

        Table result = tEnv.sqlQuery("select " +
                "    `data`['id'] id, " +
                "    `data`['paper_id'] paper_id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['score'] score, " +
                "    `data`['duration_sec'] duration_sec, " +
                "     ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='test_exam' " +
                "and (`type`='insert' or `type`='bootstrap-insert')");

        tEnv.executeSql("create table dwd_exam_user_attend_exam ( " +
                "    id string, " +
                "    paper_id string, " +
                "    user_id string, " +
                "    score string, " +
                "    duration_sec string, " +
                "    ts string " +
                ")" + SQLUtil.getKafkaSinkDDL(TOPIC_DWD_EXAM_USER_ATTEND_EXAM));

        result.executeInsert("dwd_exam_user_attend_exam");
    }
}
